package com.pai4j.connect.messagequeue.consumer.service;

import com.pai4j.common.constants.ConnectConstants;
import com.pai4j.common.enums.MessageTypeEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.connect.messagequeue.util.RedisMessageUtil;
import com.pai4j.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
import com.pai4j.domain.vo.request.mess.ChatMessageRequestVO;
import com.pai4j.domain.vo.request.mess.MessageNotifyVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Set;

/**
 * 互动通知消费服务
 * 专门处理用户互动行为产生的通知消息
 *
 * @author: CYM-pai
 * @date: 2025/01/28
 */
@Slf4j
@Service
public class InteractionNotificationConsumeService {

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;
    
    /**
     * 消费互动通知消息
     *
     * @param message 原始消息JSON字符串
     */
    public void consume(String message) {
        try {
            // 预处理消息，验证格式
            if (!RedisMessageUtil.isValidMessageFormat(message, "receiverId", "senderId", "message")) {
                log.warn("接收到格式无效的互动通知消息，跳过处理: {}", 
                        message.length() > 100 ? message.substring(0, 100) + "..." : message);
                return;
            }

            // 反序列化消息对象
            MessageNotifyVO messageNotify = JsonUtil.fromJson(message, MessageNotifyVO.class);
            if (messageNotify == null) {
                log.warn("消息反序列化失败，messageNotify为null: {}", 
                        message.length() > 100 ? message.substring(0, 100) + "..." : message);
                return;
            }

            if (StringUtils.isBlank(messageNotify.getReceiverId())) {
                log.warn("接收者ID为空，跳过处理: senderId={}, message={}", 
                        messageNotify.getSenderId(), messageNotify.getMessage());
                return;
            }

            log.info("处理互动通知消息: senderId={}, receiverId={}, message={}, content={}",
                    messageNotify.getSenderId(),
                    messageNotify.getReceiverId(),
                    messageNotify.getMessage(),
                    messageNotify.getContent());

            // 先持久化消息，确保消息不丢失且用户在不同设备都能查看
            persistentInteractionNotification(messageNotify);

            // 然后尝试通过WebSocket推送给在线用户
            pushInteractionNotificationToUser(messageNotify.getReceiverId(), messageNotify);

        } catch (Exception e) {
            log.error("处理互动通知消息失败: {}, 错误: {}", 
                    message.length() > 200 ? message.substring(0, 200) + "..." : message, e.getMessage(), e);
        }
    }

    /**
     * 通过WebSocket推送互动通知给用户
     *
     * @param receiverId 接收者ID
     * @param eventMessage
     */
    private void pushInteractionNotificationToUser(String receiverId, MessageNotifyVO eventMessage) {
        /**
         * 支持多端消息推送
         */
        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        boolean pushed = false;

        for (String bizCode : bizCodes) {
            WebSocketSession wsSession = ChatWebSocketEndPointServletHandler.getSession(bizCode, receiverId);
            /**
             * 以下两种场景不直接推送：
             *
             * 1. 在分布式场景下，每个用户的ws链接可能会落在不同的节点，所以广播模式下有可能当前节点拿不到接受人的ws session
             * 2. wssession 是一个有状态的长链接（tomcat来管理状态的），可能存在session状态异常
             */
            if (wsSession == null || !wsSession.isOpen()) {
                continue;
            }

            WebSocketMessage<?> wsMessage = new TextMessage(JsonUtil.toJsonString(eventMessage));
            try {
                wsSession.sendMessage(wsMessage);
                pushed = true;
                log.info("互动通知消息推送成功: bizCode={}, receiverId={}, wsMessage={}", bizCode, receiverId,wsMessage);
            } catch (IOException e) {
                log.error("互动通知WebSocket消息推送失败: bizCode={}, receiverId={}", bizCode, receiverId, e);
            }
        }

        if (!pushed) {
            log.info("用户不在线，互动通知消息仅通过持久化方式通知: receiverId={}", receiverId);
        } else {
            log.info("用户在线，互动通知消息已实时推送并持久化: receiverId={}", receiverId);
        }
    }

    /**
     * 持久化互动通知消息
     * 无论用户是否在线，都将互动通知消息持久化到数据库，确保用户在不同设备都能查看消息历史
     * 
     * 【关键修复】完整复制MessageNotifyVO的所有字段到ChatMessageRequestVO：
     * - 基础字段：senderId、receiverId、message、time等
     * - 扩展字段：messageType、content、commentId、replyId、resourceId等
     * 
     * 注意：MessageNotifyVO和ChatMessageRequestVO都继承自MessageBaseVO，
     * 但MessageQueueProducer的泛型参数限定为ChatMessageRequestVO，
     * 所以需要完整复制所有字段，不能直接发送MessageNotifyVO对象。
     *
     * @param messageNotify 互动通知消息
     */
    private void persistentInteractionNotification(MessageNotifyVO messageNotify) {
        try {
            // 创建ChatMessageRequestVO对象并完整复制所有字段
            ChatMessageRequestVO persistentMessage = new ChatMessageRequestVO();
            
            // === 复制基础字段（来自MessageBaseVO） ===
            persistentMessage.setType(MessageTypeEnum.INTERACTION.getType());
            persistentMessage.setSenderId(messageNotify.getSenderId());
            persistentMessage.setReceiverId(messageNotify.getReceiverId());
            persistentMessage.setProxySenderId(messageNotify.getProxySenderId());
            persistentMessage.setSessionId(messageNotify.getSessionId());
            persistentMessage.setSender(messageNotify.getSender());
            persistentMessage.setProxySender(messageNotify.getProxySender());
            persistentMessage.setMessage(messageNotify.getMessage());
            persistentMessage.setImage(messageNotify.getImage());
            persistentMessage.setTime(messageNotify.getTime());
            persistentMessage.setMessageExt(messageNotify.getMessageExt());
            
            // === 复制MessageNotifyVO的扩展字段（通过messageExt传递） ===
            // 将MessageNotifyVO的扩展字段封装到messageExt中，以便在持久化时提取
            // 这样可以绕过ChatMessageRequestVO类型限制，同时保留所有扩展数据
            java.util.Map<String, Object> extData = new java.util.HashMap<>();
            extData.put("messageType", messageNotify.getMessageType());
            extData.put("content", messageNotify.getContent());
            extData.put("commentId", messageNotify.getCommentId());
            extData.put("replyId", messageNotify.getReplyId());
            extData.put("resourceId", messageNotify.getResourceId());
            extData.put("resourceType", messageNotify.getResourceType());
            extData.put("resourceTitle", messageNotify.getResourceTitle());
            extData.put("resourceImage", messageNotify.getResourceImage());
            extData.put("resourceUrl", messageNotify.getResourceUrl());
            
            // 设置扩展数据
            persistentMessage.setMessageExt(extData);

            // 发送到消息持久化队列
            messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_INTERACTION_MESSAGE, persistentMessage);

            log.info("互动通知消息已发送到持久化队列: senderId={}, receiverId={}, messageType={}, content={}, commentId={}, message={}",
                    messageNotify.getSenderId(),
                    messageNotify.getReceiverId(),
                    messageNotify.getMessageType(),
                    messageNotify.getContent(),
                    messageNotify.getCommentId(),
                    messageNotify.getMessage());

        } catch (Exception e) {
            // 持久化失败不影响主业务，只记录日志
            log.error("互动通知消息持久化失败: senderId={}, receiverId={}, messageType={}, content={}, message={}",
                    messageNotify.getSenderId(),
                    messageNotify.getReceiverId(),
                    messageNotify.getMessageType(),
                    messageNotify.getContent(),
                    messageNotify.getMessage(), e);
        }
    }

}
